package com.atguigu.userprofile.app

import com.atguigu.userprofile.common.bean.{TagInfo, TaskInfo, TaskTagRule}
import com.atguigu.userprofile.common.const.ConstCode
import com.atguigu.userprofile.common.dao.{TagInfoDAO, TaskInfoDAO, TaskTagRuleDAO}
import com.atguigu.userprofile.common.util.{MyPropertiesUtil, MySQLUtil}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import java.util.Properties

object TaskSQL {



  //   1 自动创建  2   在写入数据前创建
  //3   create table  表名？  字段名？ 字段类型？ 库？ hive？ 分区？
  // 压缩方式？ 存储结构？   位置 ？
  // 表名 ：  标签编码
  //字段名： uid  用户id  String   ,tag_value 标签的值   标签定义里 有 标签值类型 tag_value_type
  //库：  user_profile_1018
  //分区：dt  日期
  // 不做压缩   只用文本    因为相对数仓数据量 小很多
  // 位置： 指定目录 一个表一个目录 目录名同表名
  //
  //要实现动态建表，需要提前从mysql中查询标签 和任务的相关定义！！！
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("task_sql_app")//.setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()

    //  程序怎么知道现在要计算哪个标签？
     // 平台会把每个任务的任务id ，放到json中， 远程提交器会把该任务id作为 spark-submit的参数 传递到主函数中
    //所以 咱们可以通过取得主函数的args的第一个参数作为task_id 来查询对应的标签和定义
     val taskId: String = args(0)
     val busiDate: String = args(1)

    //从mysql中查询3个 tagInfo(定义）  taskInfo（sql）  task_tag_rule(值映射)
    val tagInfo: TagInfo = TagInfoDAO.getTagInfoByTaskId(taskId)

    val taskInfo: TaskInfo = TaskInfoDAO.getTaskInfoById(taskId)

    val taskTagRuleList: List[TaskTagRule] = TaskTagRuleDAO.getTaskTagRule(taskId)

    val properties: Properties = MyPropertiesUtil.load("config.properties")
    val hdfsPath: String = properties.getProperty("hdfs-store.path")
    val upDbName: String = properties.getProperty("user-profile.dbname")
    val dwDbName: String = properties.getProperty("data-warehouse.dbname")

    val tableName=tagInfo.tagCode.toLowerCase

    val tagValueType : String = tagInfo.tagValueType match {
      case ConstCode.TAG_VALUE_TYPE_LONG => "bigint"
      case ConstCode.TAG_VALUE_TYPE_DECIMAL => "DECIMAL(16,2)"
      case ConstCode.TAG_VALUE_TYPE_STRING => "string"
      case ConstCode.TAG_VALUE_TYPE_DATE => "string"
    }


    //创建标签结果表
    //create table  $dbName.$table_name
    // ( uid string ,tag_value $tag_value_type )
    // partitioned by (dt string)
    // ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
    // location $hdfsPath/$dbName/$table_name

    val createTableSQL=
      s"""
         |     create table if not exists $upDbName.$tableName
         |    ( uid string ,tag_value $tagValueType )
         |     partitioned by (dt string)
         |      ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
         |     location '$hdfsPath/$upDbName/$tableName'
         |
         |""".stripMargin

    println(createTableSQL)
    sparkSession.sql(createTableSQL)

    // 把数据写入目标表   //组合一个insert  select 语句实现跨库的读取和写入
    var taskSQL  = taskInfo.taskSql

    //作业1-2: 把$dt 替换为业务日期
    taskSQL = taskSQL.replace("$dt", busiDate)

    //作业1-1: 如果有四级标签则用case when进行映射,如果没有query_value 就直接作为tag_value
    var queryValueSQL=""
    if(taskTagRuleList.size>0){
      val whenThenSQLList: List[String] = taskTagRuleList.map(taskTagRule => s" when '${taskTagRule.queryValue}' then  '${taskTagRule.subTagValue}'")
      val whenThenSQL: String = whenThenSQLList.mkString(" ")

       queryValueSQL=s"case query_value $whenThenSQL end   as tag_value "

    }else{
      queryValueSQL=" query_value  as tag_value"
    }

    //selec uid, case xxx when xxx then xx when xxx then xx end   as tag_value, $busiDate  from (
    //select  id as  uid ,
    // if(gender is null ,'U',gender) as query_value
    // from dim_user_zip  where dt='9999-12-31'
    //  )
    val selectSQL= s"""   |  select uid,  $queryValueSQL  from (
            |     $taskSQL  ) tg""".stripMargin

    val useDefaultTableSQL=s"use $dwDbName"
    val insertSQL=s" insert overwrite table  $upDbName.$tableName  partition (dt ='$busiDate') $selectSQL"

    println(useDefaultTableSQL)
    sparkSession.sql(useDefaultTableSQL)
    println(insertSQL)
    sparkSession.sql(insertSQL)



  }

}
